package amq.demo;

import java.util.Date;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import lang.JLTool;

import org.apache.activemq.ActiveMQConnectionFactory;

import amq.demo.thds.SQThd;
import amq.utils.GlobalAMQ;
import amq.utils.enums.ELangCode;

public class SentConsumer3 implements Runnable{
	public boolean isStop = false;
	String brokerURI;
	JLTool jlt;
	ConnectionFactory connectionFactory;  
    Connection connection = null;  
    Session session = null;
    SQThd sqThd = null;
	
	public SentConsumer3(String bui)
	{
		this.brokerURI = bui;
	}
	
	protected void sayHi()
	{
		MessageProducer producer = null;		
		try
		{			
			Destination destination = session.createQueue(GlobalAMQ.SayHiTopic);
			producer = session.createProducer(destination);
	        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	        TextMessage message = session.createTextMessage(String.format("%s:hi", this.getClass().getName()));	   
	        message.setStringProperty("Topic", GlobalAMQ.QueryTopic);
			producer.send(message);
			System.out.printf("\t[%s] Say Hi...\n", this.getClass().getName());
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		finally
		{
			if(producer!=null) 
			{
				try{producer.close();}catch(Exception e){}
			}
		}
	}
	
	protected void sayBye()
	{
		MessageProducer producer = null;		
		try
		{			
			Destination destination = session.createQueue(GlobalAMQ.SayHiTopic);
			producer = session.createProducer(destination);
	        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	        TextMessage message = session.createTextMessage(String.format("%s:bye", this.getClass().getName()));
	        message.setStringProperty("Topic", GlobalAMQ.QueryTopic);
			producer.send(message);
			System.out.printf("\t[%s] Say Bye...\n", this.getClass().getName());
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		finally
		{
			if(producer!=null) 
			{
				try{producer.close();}catch(Exception e){}
			}
		}
	}
	
	protected void _response(String dest, String tid)
	{
		MessageProducer producer = null;		
		try
		{			
			Destination destination = session.createQueue(dest);
			producer = session.createProducer(destination);
	        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
	        TextMessage message = null;
	        
			// tuple(Rule_ID,Category,Hit_Part,Correction, IRule)
	        message = session.createTextMessage("Fail");
			message.setBooleanProperty("Correct", false);
			message.setLongProperty("TimeStamp", new Date().getTime());
			message.setStringProperty("ServiceName", "AlwaysFail");
			message.setStringProperty("TID", tid);
			message.setStringProperty("Category", "Category");			
			message.setStringProperty("Reason", "Reason");
			message.setStringProperty("Suggestion", "Suggestion");
			producer.send(message);
			System.out.printf("\t[%s] Send answer back!\n", this.getClass().getName());
		}
		catch(JMSException e)
		{
			e.printStackTrace();
		}
		finally
		{
			if(producer!=null) 
			{
				try{producer.close();}catch(Exception e){}
			}
		}
	}
	
	@Override
	public void run() {				  
        Destination destination;  
        MessageConsumer consumer = null;  
        TextMessage message = null;  
        boolean useTransaction = false;  
        try {         
        	// 0) Start Service Query Thread
        	sqThd = new SQThd(brokerURI, this.getClass().getName(), GlobalAMQ.QueryTopic, ELangCode.SimpleChinese);
        	new Thread(sqThd).start();
        	
        	// 1) Start AMQ Producer
            connectionFactory = new ActiveMQConnectionFactory(brokerURI);  
            connection = connectionFactory.createConnection();  
            connection.start();  
            session = connection.createSession(useTransaction,  
                    Session.AUTO_ACKNOWLEDGE);  
            destination = session.createTopic(SentProducer.QueryTopic);
            consumer = session.createConsumer(destination);  
            System.out.printf("\t[%s] Consumer starting...\n", this.getClass().getName());
            sayHi();
            while(!isStop)
            {
            	message = (TextMessage) consumer.receive(1000);  
                if(message!=null) {
                	String line = message.getText();
                	System.out.printf("\t[%s] Received message:\n%s\n", this.getClass().getName(), message);                	
                	System.out.printf("\t\tUID: %s\n", message.getStringProperty("UID"));
                	System.out.printf("\t\tTID: %s\n", message.getStringProperty("TID"));
                	System.out.printf("\t\tReplyTo: %s\n", message.getStringProperty("ReplyTo"));
                	System.out.printf("\t\tSENT: %s\n", line);                	
        			_response(message.getStringProperty("ReplyTo"), message.getStringProperty("TID"));
                }
                
            }
            System.out.printf("\t[%s] Bye!\n", this.getClass().getName()); 
        } catch (JMSException jmsEx) {  
  
        } finally {  
            try{
            	sayBye();
            	consumer.close();  
                session.close();  
                connection.close();  
                sqThd.isStop = true;
                sqThd = null;
            } catch(Exception e){}
        }  
		
	}

	
	/**
	 * BD: Sentence Topic Consumer2 - Always response pass
	 * @param args
	 */
	public static void main(String[] args) throws Exception{
		if(args.length==0)
		{
			System.out.printf("\t[Info] Please give Connector URI! (ex 'tcp://localhost:61616')\n");
			return;
		}
		SentConsumer3 sc3 = new SentConsumer3(args[0]);
		new Thread(sc3).start();
		
		System.out.printf("\t[Info] Press any key to stop...\n");
		System.in.read();
		sc3.isStop = true;
		System.out.printf("\t[Info] Exit %s...\n", sc3.getClass().getName());
	}
}
